package com.dongdongshop.mq;

import cn.hutool.json.JSONUtil;
import com.dongdongshop.controller.FreemarkerController;
import com.dongdongshop.service.FreemarkerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
//消费者需要用注解配置topic主题 和消费者的组
@RocketMQMessageListener(topic = "ggw", consumerGroup = "ggw")
//消费者需要继承接口RocketMQListener<>可以通过泛型接受消费消息
//ack重试机制要继承RocketMQPushConsumerLifecycleListener
public class Consume implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private FreemarkerService freemarkerService;

    //继承RocketMQListener的方法
    @Override
    public void onMessage(String s) {
        //继承RocketMQPushConsumerLifecycleListener的接口后这个方法就作废了
        System.out.println("接收到信息 : " + s);
    }

    //继承RocketMQPushConsumerLifecycleListener后的方法
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        //通过方法创建内部类内部类中写业务
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //list就是消费消息
                //单条消息直接下表0获取对象,可以通过对象获取更多属性
                MessageExt messageExt = list.get(0);
                //new string接受消息内容 如果不转string会是一个二进制字符串
                String msg = new String(messageExt.getBody());
                //获取消息id
                String msgId = messageExt.getMsgId();
                //获取消息重试次数
                int reconsumeTimes = messageExt.getReconsumeTimes();
                //ctrl+alt+t  生成try
                try {
                    //查看redis缓存中有没有msgid,如果有就不再消费,返回true代表已经有msgid
                    Boolean mqmessage = redisTemplate.boundHashOps("MQMESSAGE").hasKey(msgId);
                    if (!mqmessage) {
                        System.out.println("ack消息:" + msg);
                        List<Long> longs = JSONUtil.toList(msg, Long.class);
                        for (int i = 0; i < longs.size(); i++) {
                            freemarkerService.createPage(longs.get(i));
                        }
                        //int i = 1 / 0;
                        redisTemplate.boundHashOps("MQMESSAGE").put(msgId, 1);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    if (reconsumeTimes >= 3) {//如果重试次数大于3,没必要继续尝试返回成功,并且将消息放入死信队列
                        System.out.println("放入死信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    System.out.println("ACK消息接受失败,请稍后重试");
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//发生异常返回重新消费
                }
                //消费成功返回success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
